﻿#ifndef UDPCLIENT_H
#define UDPCLIENT_H
#include <QThread>
#include <QMutex>
#include <atomic>
#include <librdkafka/rdkafkacpp.h>


class KafkaClient : public QThread
{
    Q_OBJECT

public:
    explicit KafkaClient(std::string, QObject *parent = nullptr);
    ~KafkaClient();
    void notify_thread_exit();

signals:
	void kafkaMessageReceived(const QString &msg, const std::string& key, const QDateTime& recv_time);
    void kafkaFailedMsg(const QString& msg);

protected:
    //重写线程执行函数
    void run() override;
    void msg_consume(RdKafka::Message* message, void* opaque);

private:
    std::string m_kafkaIp;
    RdKafka::KafkaConsumer *consumer;
    RdKafka::Conf *global_conf;
    RdKafka::Conf *topic_conf;
    std::atomic_bool g_bUdpThreadRun;
    int64_t msg_cnt, msg_bytes;

};

#endif // UDPCLIENT_H
